package com.abyss.watermark;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

/**
 * 使用一个事件时间的 滚动窗口，并设置水印
 */
public class TumblingEventTimeWindowWithWatermarkDemo {
    public static void main(String[] args) throws Exception {
        // 1. Env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // 2. 告知Flink 我要用EventTime啦
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        // 3. Socket source, 输入数据: 单词 时间戳
        DataStreamSource<String> socketTextStream = env.socketTextStream("node1", 9999);

        // 4. 将socket数据转变为元组
        SingleOutputStreamOperator<Tuple2<String, Long>> wordWithTSDataStream = socketTextStream.map(new MapFunction<String, Tuple2<String, Long>>() {
            @Override
            public Tuple2<String, Long> map(String value) throws Exception {
                String[] ss = value.split(" ");
                return Tuple2.of(ss[0], Long.parseLong(ss[1]));
            }
        });

        // 5. 水印
        SingleOutputStreamOperator<Tuple2<String, Long>> watermarks = wordWithTSDataStream.assignTimestampsAndWatermarks(
                new BoundedOutOfOrdernessTimestampExtractor<Tuple2<String, Long>>(Time.seconds(3)) {
                    @Override
                    public long extractTimestamp(Tuple2<String, Long> element) {
                        return element.f1;  // 告知Flink 事件时间
                    }
                }
        );

        // 6. 先水印 后keyby 最后加窗口
        SingleOutputStreamOperator<Tuple2<String, Long>> result = watermarks.keyBy(0).timeWindow(Time.seconds(5)).sum(1);

        result.print();
        env.execute();
    }
}